// Copyright 2014 The mqrouter Author. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package xredismq

import (
	"context"
	"encoding/json"
	"gitee.com/go-mid/infra/xcache/xredis"
	"gitee.com/go-mid/infra/xmq/mqconf"
	"gitee.com/go-mid/infra/xtime"
	"gitee.com/go-mid/infra/xtrace"
	"github.com/go-redis/redis"
)

type WriterConfig struct {
	mqconf.MQConfig
	RedisConf xredis.RedisConfig `json:"redis_conf"`
	Topic     string             `json:"topic"`
	Model     RedisMQModel       `json:"model"`
}
type RedisWriter struct {
	redisClient *redis.Client
	config      *WriterConfig
	stop        chan struct{}
}

func NewRedisWriter(writerConfig *WriterConfig) (*RedisWriter, error) {
	var ctx = context.Background()
	redisClient, err := xredis.NewRedisClient(ctx, &writerConfig.RedisConf)
	if err != nil {
		return nil, err
	}
	client := redisClient.GetRawClient(ctx)
	redisWriter := &RedisWriter{
		redisClient: client,
		stop:        make(chan struct{}),
		config:      writerConfig,
	}

	return redisWriter, nil
}

func (m *RedisWriter) setRedisSpanTags(span xtrace.Span) {
	config := m.config
	span.SetTag(xtrace.TagSpanKind, xtrace.SpanKindProducer)
	span.SetTag(xtrace.TagMessageBusDestination, config.Topic)
	span.SetTag(xtrace.TagMessagingDestinationKind, xtrace.MessagingDestinationKindTopic)
	span.SetTag(xtrace.TagMessagingDestination, config.Topic)
}

var xaddkey = "_msg_"

func (m *RedisWriter) WriteMsg(ctx context.Context, k string, v interface{}) (string, error) {
	span := xtrace.SpanFromContext(ctx)
	if span != nil {
		m.setRedisSpanTags(span)
	}
	msg, err := json.Marshal(v)
	if err != nil {
		return "", err
	}
	st := xtime.NewTimeStat()
	switch m.config.Model {
	case RedisMQModel_PubSub:
		_, err = m.redisClient.Publish(m.config.Topic, string(msg)).Result()
	case RedisMQModel_Stream:
		_, err = m.redisClient.XAdd(&redis.XAddArgs{
			Stream: m.config.Topic,
			Values: map[string]interface{}{xaddkey: string(msg)},
		}).Result()
	}
	mqconf.StatReqDuration(ctx, m.config.Topic, "RedisWriter.WriteMessages", mqconf.TraceMessageBusTypeRedis, st.Millisecond())
	return "", err
}

func (m *RedisWriter) Close(ctx context.Context) (err error) {
	m.stop <- struct{}{}
	return m.redisClient.Close()
}
